package com.haust.consumer;

import com.haust.dao.RedisDao;
import com.haust.pojo.TbItem;
import com.haust.pojo.TbOrderItem;
import com.haust.rabbit.message.dto.HaustCreateOrderMessageBody;
import com.haust.rabbit.message.dto.HaustSendEmailMessageBody;
import com.haust.rabbit.sender.MessageSender;
import com.haust.service.CreateOrderService;
import org.springframework.amqp.rabbit.annotation.Exchange;
import org.springframework.amqp.rabbit.annotation.Queue;
import org.springframework.amqp.rabbit.annotation.QueueBinding;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 创建订单的消费者
 */
@Component
public class CreateOrderConsumer {
    @Value("${Haust.createOrder.lock}")
    private String lock;

    @Value("${Haust.frontend.details.item}")
    private String itemKeyPrefix;

    @Value("${Haust.trade.sendMessage.exchange}")
    private String sendMessageExchange;

    @Value("${Haust.trade.sendMessage.routingKey}")
    private String sendMessageRoutingKey;

    @Autowired
    private MessageSender messageSender;

    @Autowired
    private RedisDao redisDao;

    @Autowired
    private CreateOrderService createOrderService;

    final ThreadLocal<Integer> threadLocal = ThreadLocal.withInitial(() -> 0);

    /**
     * 创建订单的消费者
     * 1、获取分布式锁
     * 使用全局（创建订单）的分布式锁，把redis中的要购买的商品的库存修改
     * 释放分布式锁。当分布式事务出现回滚的时候，再将redis中要购买的商品的库存还原
     * 分布式事务要做的事情：1、新增订单表，订单项表，物流表 2、修改商品库存 3、修改（清空）redis中的购物车
     * 2、开启分布式事务
     * 3、创建订单，物流，订单数据项数据
     * 4、修改库存中的商品的库存
     * 5、修改redis中购物车数据
     * 6、提交或者回滚数据
     *
     * @param messageBody
     */
    @RabbitListener(
            bindings = {
                    @QueueBinding(
                            value = @Queue(
                                    name = "${Haust.trade.createOrder.queue}",
                                    autoDelete = "false"
                            ),
                            exchange = @Exchange(
                                    name = "${Haust.trade.createOrder.exchange}",
                                    type = "${Haust.trade.createOrder.exchangeType}"
                            ),
                            key = "${Haust.trade.createOrder.routingKey}"
                    )
            }
    )
    public void onMessage(HaustCreateOrderMessageBody messageBody) {
        createOrder(messageBody);
    }

    private Boolean createOrder(HaustCreateOrderMessageBody messageBody) {
        Boolean isLock = redisDao.setnx(lock, "", 2L, TimeUnit.SECONDS);
        if (!isLock) {
            int spinCount = threadLocal.get();
            if (spinCount == 10) {
                return false;
            }
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            spinCount++;
            threadLocal.set(spinCount);
            //如果上锁失败
            return createOrder(messageBody);
        }
        //上锁成功 ,进行分布式事务

        //1、更新 redis中的商品库存
        List<TbOrderItem> orderItems = messageBody.getTbOrderItem();
        for (TbOrderItem orderItem : orderItems) {
            TbItem tbItem = redisDao.get(itemKeyPrefix + orderItem.getItemId());
            if (tbItem.getNum() < orderItem.getNum()) {
                //清空threadLocal
                threadLocal.remove();
                //redis中商品库存小于 购物车中的该商品的数量
                return false;
            }
            tbItem.setNum(tbItem.getNum() - orderItem.getNum());
            redisDao.set(itemKeyPrefix + orderItem.getItemId(), tbItem, 7L, TimeUnit.DAYS);
        }
        //释放分布式锁
        redisDao.del(lock);


        //当消息消费失败的时候，消息会重新回退到队列中，消费者重新消费，为了避免这种情况。
        //当有分布式事务出现异常，我们抓住异常，并且进行事务的回滚。将redis中减少掉的商品的数量 给重新添加回去
        try {
            createOrderService.createOrder(messageBody);

            //创建订单成功之后，给用户发送订单成功的消息，让用户进行支付
            HaustSendEmailMessageBody sendEmailMessageBody = new HaustSendEmailMessageBody();

            sendEmailMessageBody.setTbUser(messageBody.getTbUser());
            sendEmailMessageBody.setOrderId(messageBody.getTbOrder().getOrderId());
            sendEmailMessageBody.setPayment(messageBody.getTbOrder().getPayment());


            messageSender.sendMessage(sendMessageExchange,
                    sendMessageRoutingKey,
                    sendEmailMessageBody);

        } catch (Exception e) {
            //打印错误
            e.printStackTrace();
            //重新设置 自旋次数
            threadLocal.set(0);
            //会滚商品数量
            rollbackRedisItemNum(messageBody);
            //清空threadLocal
            threadLocal.remove();
            return false;
        }

        //清空threadLocal
        threadLocal.remove();
        return true;
    }

    private Boolean rollbackRedisItemNum(HaustCreateOrderMessageBody messageBody) {
        Boolean isLock = redisDao.setnx(lock, "", 2L, TimeUnit.SECONDS);
        if (!isLock) {
            int spinCount = threadLocal.get();
            if (spinCount == 10) {
                return false;
            }
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            spinCount++;
            threadLocal.set(spinCount);
            //如果上锁失败
            return rollbackRedisItemNum(messageBody);
        }
        //上锁成功 ,进行分布式事务

        //1、更新 redis中的商品库存
        List<TbOrderItem> orderItems = messageBody.getTbOrderItem();
        for (TbOrderItem orderItem : orderItems) {
            TbItem tbItem = redisDao.get(itemKeyPrefix + orderItem.getItemId());
            tbItem.setNum(tbItem.getNum() + orderItem.getNum());
            redisDao.set(itemKeyPrefix + orderItem.getItemId(), tbItem, 7L, TimeUnit.DAYS);
        }
        //redis中商品库存还原成功，释放分布式锁
        redisDao.del(lock);

        return true;
    }
}
